import {
  Any,
  AnyObject,
  Callback,
  Predicate
} from "../types";
import { assert } from "../util";
import { Source } from "../lazy";
import { CollationSpec } from "../core";


// 异步源类型，支持异步可迭代对象和普通可迭代对象
export type AsyncSource = Source | AsyncIterable<Any> | (() => Promise<AsyncIteratorResult>) | OpenableAsyncIterable;

// 异步迭代器结果接口
export interface AsyncIteratorResult<T = Any> {
  readonly value?: T;
  readonly done: boolean;
}

/**
 * Symbol used to check if an async iterable has been opened.
 * This allows us to check the state of an OpenableAsyncIterable without
 * actually opening it.
 */
export const OpenedSymbol = Symbol.for("opened");

/**
 * An interface that extends AsyncIterable to support "opening" with conditions.
 * This is useful for data sources that can optimize their data retrieval based
 * on query conditions before actually iterating over the data.
 *
 * This interface allows data sources like database connections to receive query
 * conditions upfront, enabling them to perform more efficient data retrieval
 * rather than fetching all data and then filtering it in memory.
 */
export interface OpenableAsyncIterable extends AsyncIterable<Any> {
  /**
   * Checks if the async iterable has been opened.
   * @returns true if the iterable has been opened, false otherwise.
   */
  [OpenedSymbol]: () => boolean;

  /**
   * Opens the async iterable with the given condition.
   * This method should be called before iterating over the data to allow
   * the data source to optimize its retrieval based on the provided condition.
   *
   * @param condition - The query condition to pass to the data source.
   *                   The exact format and interpretation of this condition
   *                   depends on the specific implementation.
   * @returns A promise that resolves when the iterable has been opened.
   */
  open(condition: any): Promise<void>;

  /**
   * Creates a clone of the async iterable.
   * This is useful when the same data source needs to be queried multiple times
   * with different conditions within the same aggregation pipeline.
   *
   * @returns A new OpenableAsyncIterable instance that represents the same data source.
   */
  clone(): OpenableAsyncIterable;
}

export interface Closeable {
  close(): Promise<void>;
}

export function isCloseable(o: Any): boolean {
  return (
    !!o &&
    typeof o === "object" &&
    typeof (o as Closeable)?.close === "function"
  );
}
/**
 * Creates an async lazy iterator from the given source.
 */
export function AsyncLazy(source: AsyncSource): AsyncIterator {
  return source instanceof AsyncIterator ? source : new AsyncIterator(source);
}

/**
 * Concatenate multiple async iterators and return a new async iterator.
 *
 * @param iterators The async iterators to concatenate
 * @returns {AsyncIterator} A new async iterator
 */
export function asyncConcat(...iterators: AsyncIterator[]): AsyncIterator {
  let index = 0;
  return AsyncLazy(async () => {
    while (index < iterators.length) {
      const o = await iterators[index].next();
      if (!o.done) return { value: o.value, done: false };
      index++;
    }
    return { done: true, value: undefined };
  });
}

/**
 * Checks whether the given object is compatible with a generator i.e Object{next:Function}
 * @param {*} o An object
 */
function isAsyncGenerator(o: Any) {
  return (
    !!o && typeof o === "object" && typeof (o as AnyObject)?.next === "function"
  );
}

function isAsyncIterable(o: Any) {
  return (
    !!o &&
    (typeof o === "object" || typeof o === "function") &&
    typeof o[Symbol.asyncIterator] === "function"
  );
}

export function isOpenableAsyncIterable(o: Any): boolean {
  return (
    isAsyncIterable(o) &&
    typeof o[OpenedSymbol] === "function"
  );
}

interface AsyncIteratee {
  op: "map" | "filter";
  fn: Callback<Promise<Any> | Any>;
}

// export type AsyncSource = Source | AsyncIterable<Any> | (() => Promise<AsyncIteratorResult>);

/**
 * An async lazy collection iterator yields a single value at a time upon request.
 *
 * This class supports OpenableAsyncIterable sources, which allows it to pass query
 * conditions to the underlying data source before iteration begins. This enables
 * more efficient data retrieval from sources that can optimize based on query conditions,
 * such as database connections.
 */
export class AsyncIterator {
  #iteratees: AsyncIteratee[] = [];
  #buffer: Any[] = [];
  #getNext: () => Promise<AsyncIteratorResult>;
  #done = false;
  #openableIter: OpenableAsyncIterable | null = null;
  _matchCondition: Array<any> = [];

  /**
   * Checks if the underlying source supports open operation.
   * This is true when the source implements the OpenableAsyncIterable interface.
   *
   * @returns true if the source supports open operation, false otherwise.
   */
  supportsOpen(): boolean {
    return !!this.#openableIter;
  }

  /**
   * Checks if the underlying source is opened.
   *
   * @returns true if the source is opened, false otherwise.
   */
  opened(): boolean {
    return this.#openableIter?.[OpenedSymbol]() ?? false;
  }

  /**
   * Adds a condition to be passed to the underlying OpenableAsyncIterable when it's opened.
   * Multiple conditions can be added, and they will all be passed to the open() method.
   *
   * @param condition - The condition to add.
   */
  addCondition(c: any) {
    if (this.supportsOpen()) {
      this._matchCondition.push(c);
    }
  }

  /**
   * Opens the underlying source with given condition.
   * If the source doesn't support open operation or is already opened, this method does nothing.
   *
   * @param condition - Query condition to pass to the source.
   * @returns A promise that resolves when the source has been opened.
   */
  async open(condition: any): Promise<void> {
    if (this.opened()) return Promise.resolve();
    if (!this.#openableIter) return Promise.resolve();
    return this.#openableIter.open(condition);
  }

  constructor(source: AsyncSource) {
    // 处理不同类型的源
    let asyncIter: any = null;

    if (isAsyncIterable(source)) {
      asyncIter = (source as AsyncIterable<Any>)[Symbol.asyncIterator]();
      // Check if it's an openable async iterable
      if (isOpenableAsyncIterable(source)) {
        this.#openableIter = source as OpenableAsyncIterable;
      }
    } else if (isAsyncGenerator(source)) {
      asyncIter = source;
    } else if (typeof source === "function") {
      // 函数形式的源
      asyncIter = { next: source };
    } else if (Symbol.iterator in Object(source)) {
      // 如果是同步可迭代对象，包装成异步的
      const syncIter = source[Symbol.iterator]();
      asyncIter = {
        next: async () => {
          const result = syncIter.next();
          return Promise.resolve(result);
        }
      };
    }

    assert(
      !!asyncIter,
      `AsyncIterator must be initialized with an async iterable, sync iterable or function.`
    );

    // index of successfully transformed and yielded item
    let index = -1;

    // create function to yield the next transformed value
    this.#getNext = async () => {
      if(this.supportsOpen() && !this.opened()){
        await this.open(this._matchCondition)
      }
      while (true) {
        const current = await asyncIter.next();
        if (current.done) break;

        let value = current.value;
        index++;

        // Apply all iteratees
        let shouldInclude = true;
        for (const { op, fn } of this.#iteratees) {
          const res = await fn(value, index);
          if (op === "map") {
            value = res;
          } else if (op === "filter") {
            shouldInclude = !!res;
            if (!shouldInclude) break;
          }
        }

        if (shouldInclude) {
          return { value, done: false };
        }
      }
      if(isCloseable(source)) {
        await (source as unknown as Closeable).close();
      }
      return { done: true };
    };
  }

  /**
   * Add an iteratee to this async lazy sequence
   */
  private push(op: "map" | "filter", fn: Callback<Promise<Any> | Any>): AsyncIterator {
    this.#iteratees.push({ op, fn });
    return this;
  }

  async next<T = Any>(): Promise<IteratorResult<T, any>> {
    const result = await this.#getNext();
    if (result.done) {
      return { done: true, value: undefined } as IteratorResult<T, any>;
    }
    return { done: false, value: result.value } as IteratorResult<T, any>;
  }

  // Iteratees methods

  /**
   * Transform each item in the sequence to a new value
   * @param {Function} f
   */
  map<T = Any>(f: Callback<Promise<T> | T>): AsyncIterator {
    return this.push("map", f);
  }

  /**
   * Select only items matching the given predicate
   * @param {Function} f
   */
  filter<T = Any>(f: Predicate<T> | ((value: T, index: number) => Promise<boolean>)): AsyncIterator {
    return this.push("filter", f as Callback<T>);
  }
  take(n: number): AsyncIterator {
    return n >= 0 ? this.filter((_: Any) => { return !(n === 0 || n-- === 0)}) : this;
  }

  /**
   * Returns a new async lazy object with results of the transformation
   * The entire sequence is realized.
   *
   * @param {Callback<AsyncSource, Promise<Any[]>>} fn Transform function of type (Array) => (Any)
   */
  async transform(fn: Callback<AsyncSource|Promise<AsyncSource>, Any[]>): Promise<AsyncIterator> {
    const self = this;
    let iter: AsyncIterator;
    return AsyncLazy(async () => {
      if (!iter) {
        const v=await self.value()
        const values = await fn(v);
        iter = AsyncLazy(values);
      }
      const result = await iter.next();
      return { value: result.value, done: result.done };
    });
  }

  /**
   * Retrieves all remaining values from the async lazy evaluation and returns them as an array.
   * This method processes the underlying iterator until it is exhausted, storing the results
   * in an internal buffer to ensure subsequent calls return the same data.
   */
  async value<T>(): Promise<T[]> {
    if (!this.#done) {
      while (true) {
        const { done, value } = await this.#getNext();
        if (done) break;
        this.#buffer.push(value);
      }
      this.#done = true;
    }
    return this.#buffer as T[];
  }

  /**
   * Execute the callback for each value.
   * @param f The callback function.
   * @returns {Boolean} Returns false if the callback returned false to break the loop, otherwise true.
   */
  async each<T = Any>(f: Callback<T, Promise<boolean> | boolean>): Promise<boolean> {
    while (true) {
      const o = await this.next();
      if (o.done) break;
      const result = await f(o.value);
      if (result === false) return false;
    }
    return true;
  }

  /**
   * Returns the reduction of sequence according the reducing function
   *
   * @param f The reducing function
   * @param initialValue The initial value
   */
  async reduce<T = Any>(
    f: (accumulator: T, currentValue: T) => Promise<T> | T,
    initialValue?: T
  ): Promise<T> {
    let o = await this.next();

    if (initialValue === undefined && !o.done) {
      initialValue = o.value as T;
      o = await this.next();
    }

    while (!o.done) {
      initialValue = await f(initialValue, o.value as T);
      o = await this.next();
    }

    return initialValue;
  }

  /**
   * Returns the number of matched items in the sequence
   */
  async size(): Promise<number> {
    return this.reduce(
      async (acc: number) => ++acc,
      0
    );
  }

  [Symbol.asyncIterator](): AsyncIterator {
    return this;
  }
}